package com.workerthread;

import com.TestByteBuffer;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author 罗俊华
 * @date 2021/8/24 - 2:37 下午
 */
@Slf4j
public class WorkerTask implements Runnable {

    /**
     * 阻塞队列用于进程间（boss进程与当前worker进程）通信
     */
    private final Queue<Runnable> queue = new ConcurrentLinkedQueue<>();

    private final Selector selector;

    private final Thread workerThread;

    private boolean started;

    @Getter
    private final String workerName;

    public WorkerTask(String workerName) {
        Selector selector1;
        try {
            selector1 = Selector.open();
        } catch (IOException e) {
            selector1 = null;
            e.printStackTrace();
        }

        this.selector = selector1;
        this.workerThread = new Thread(this, workerName);
        this.workerName = workerName;
    }


    @Override
    public void run() {

        while (true) {

            try {

                log.debug("线程【{}】等待 selector 的事件触发", Thread.currentThread().getName());

                this.selector.select();

                Runnable registerTask = this.queue.poll();

                if (registerTask != null) {
                    // 唤醒之后，可以自己把待注册的channel，注册到 selector 中，而不产生阻塞
                    log.debug("阻塞队列中有需要注册到selector的channel，现执行注册");
                    registerTask.run();
                }

                Iterator<SelectionKey> selectionKeyIterator = this.selector.selectedKeys().iterator();

                while (selectionKeyIterator.hasNext()) {

                    SelectionKey selectionKey = selectionKeyIterator.next();
                    selectionKeyIterator.remove();

                    if (selectionKey.isReadable()) {

                        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();

                        ByteBuffer readByteBuffer = (ByteBuffer) selectionKey.attachment();

                        try {
                            int read = socketChannel.read(readByteBuffer);

                            log.debug("线程【{}】触发了可写事件，本次读取【{}】字节", Thread.currentThread().getName(), read);

                            TestByteBuffer.split(readByteBuffer);

                            if (read == -1) {
                                log.debug("线程【{}】监听到数据已读完", Thread.currentThread().getName());
                                selectionKey.cancel();
                            }
                        } catch (IOException e) {
                            log.info("线程【{}】监听到client关闭了连接", Thread.currentThread().getName());
                            e.printStackTrace();
                            selectionKey.cancel();

                        }

                    }

                }


            } catch (IOException e) {
                e.printStackTrace();
            }


        }

    }

    public Runnable register(SocketChannel socketChannel) throws IOException {

        socketChannel.configureBlocking(false);

        /**
         * 将 boss 进程的 channel 注册到 工作者线程 worker 的 selector 中的动作添加到阻塞队列中
         * */
        this.queue.add(() -> {

            try {
                socketChannel.register(this.selector, SelectionKey.OP_READ, ByteBuffer.allocate(8));
            } catch (ClosedChannelException e) {
                e.printStackTrace();
            }
        });

        // 将 worker 线程从 selector.select() 阻塞方法中唤醒，让其有机会继续往下执行，由此能从阻塞队列中拿到任务 注册 channel 到 selector 中
        log.debug("准备唤醒【{}】以注册channel",this.workerName);
        this.selector.wakeup();


        if (!this.started) {
            this.workerThread.start();
            this.started = true;
            log.debug("线程【{}】给未启动的【{}】线程指定了第一个任务，现在启动【{}】", Thread.currentThread().getName(), this.workerName, this.workerName);
        }

        return this;
    }
}
